//
// Copyright 2011 Aster Data

import java.io.File;
import java.util.ArrayList;
import java.util.List;

import org.apache.lucene.analysis.Analyzer;
import org.apache.lucene.analysis.SimpleAnalyzer;
import org.apache.lucene.analysis.standard.StandardAnalyzer;
import org.apache.lucene.document.Document;
import org.apache.lucene.queryParser.QueryParser;
import org.apache.lucene.search.IndexSearcher;
import org.apache.lucene.search.Query;
import org.apache.lucene.search.ScoreDoc;
import org.apache.lucene.search.TopScoreDocCollector;
import org.apache.lucene.store.Directory;
import org.apache.lucene.store.FSDirectory;
import org.apache.lucene.util.Version;

import com.asterdata.ncluster.sqlmr.ArgumentClause;
import com.asterdata.ncluster.sqlmr.ClientVisibleException;
import com.asterdata.ncluster.sqlmr.Drainable;
import com.asterdata.ncluster.sqlmr.OutputInfo;
import com.asterdata.ncluster.sqlmr.RowFunction;
import com.asterdata.ncluster.sqlmr.RuntimeContract;
import com.asterdata.ncluster.sqlmr.TaskInfo;
import com.asterdata.ncluster.sqlmr.data.ColumnDefinition;
import com.asterdata.ncluster.sqlmr.data.RowEmitter;
import com.asterdata.ncluster.sqlmr.data.RowIterator;
import com.asterdata.ncluster.sqlmr.data.SqlType;

public final class lucene_search implements RowFunction, Drainable {
    //
    // These member variables will be populated with the values
    // of the argument clauses passed to your SQL-MR function.
    //

    private String indexdirArgument = "";
    private String analyzerArgument = "";
    private String queryArgument = "";
    private int topnArgument = 1;
    private double fuzzythresholdArgument = 0;

    //
    // The constructor establishes the RuntimeContract between
    // your function and nCluster. During query planning, your
    // function will constructed on a single node. During query
    // execution, it will be constructed and run on one or more
    // nodes.
    //

    public lucene_search(RuntimeContract contract) {

        //
        // Construct the output schema.
        //

        List<ColumnDefinition> outputColumns = new ArrayList<ColumnDefinition>();

        outputColumns.add(new ColumnDefinition("docid", SqlType.getType("character varying")));
        outputColumns.add(new ColumnDefinition("score", SqlType.getType("double precision")));

        //
        // Read argument clauses into appropriate member variables.
        //

        ArgumentClause tmpAc;
        String tmpValue;

        //
        // Read argument clause 'indexdir'
        //

        tmpAc = contract.useArgumentClause("indexdir");

        tmpValue = tmpAc.getSingleValue();
        indexdirArgument = tmpValue;

        //
        // Read argument clause 'analyzer'
        //

        tmpAc = contract.useArgumentClause("analyzer");

        tmpValue = tmpAc.getSingleValue();
        analyzerArgument = tmpValue;

        //
        // Read argument clause 'query'
        //

        tmpAc = contract.useArgumentClause("query");

        tmpValue = tmpAc.getSingleValue();
        queryArgument = tmpValue;

        // argument topn
        tmpAc = contract.useArgumentClause("topn");
        tmpValue = tmpAc.getSingleValue();
        topnArgument = Integer.parseInt(tmpValue);

        //
        // Add any other function-specific initialization of member variables
        // or other checks here. Throw a ClientVisibleException to signal
        // an error to the client.
        //

        //
        // Complete the contract
        //

        contract.setOutputInfo(new OutputInfo(outputColumns));
        contract.complete();
    }

    public void operateOnSomeRows(RowIterator inputIterator, RowEmitter outputEmitter) {

    }

    public void drainOutputRows(RowEmitter outputEmitter) {
        //
        // This method will be called exactly one time per instance of this
        // SQL-MR function (even if there are no inputs rows or partitions).
        // Use it to output any final rows.
        //

        int taskindex = TaskInfo.getTaskIndex();
        File indexdir = new File(indexdirArgument + "/" + taskindex);

        if (indexdir.exists()) { // it's possible that index does not exist on a particular vworker

            // initialize analyze
            Analyzer analyzer = null;
            if (analyzerArgument.equals("simple"))
                analyzer = new SimpleAnalyzer();
            else if (analyzerArgument.equals("standard"))
                analyzer = new StandardAnalyzer(Version.LUCENE_30);
            else
                throw new ClientVisibleException("analyzer parameter value \"" + analyzerArgument
                        + "\" is not valid. supported values are simple/standard.");

            // initialize lucene
            try {
                Directory dir = FSDirectory.open(indexdir);
                IndexSearcher isearcher = new IndexSearcher(dir);

                // Parse a simple query that searches for "text":
                QueryParser parser = new QueryParser(Version.LUCENE_30, "text", analyzer);
                Query query = parser.parse(queryArgument);

                TopScoreDocCollector collector = TopScoreDocCollector.create(topnArgument, true);
                isearcher.search(query, collector);

                // Iterate through the results:
                ScoreDoc[] hits = collector.topDocs().scoreDocs;
                for (int i = 0; i < hits.length; i++) {
                    Document hitDoc = isearcher.doc(hits[i].doc);
                    outputEmitter.addString(hitDoc.get("docid"));
                    outputEmitter.addDouble(hits[i].score);
                    outputEmitter.emitRow();
                }

                isearcher.close();
                dir.close();

            } catch (Exception e) {
                StackTraceElement[] strace = e.getStackTrace();
                String msg = "";
                for (int i = 0; i < strace.length; i++)
                    msg += strace[i].toString() + "\n";
                throw new ClientVisibleException("lucene search exception: " + e.toString() + "\n" + msg);
            }
        }
    }
}
